package com.zlc.common.util;

import cn.hutool.core.lang.UUID;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Supplier;

@Service
@Slf4j
public class FluxUtil {

    @Resource(name = "webFluxExecutor")
    private ThreadPoolTaskExecutor executor;

    private static final AtomicLong current = new AtomicLong(0);

    /**
     * 创建一个默认的sse
     * 默认 3 秒
     * @return sse
     */
    public Flux<String> createSSE(Supplier<String> success, Supplier<String> error) {
        return createSSE(success, error, 3);
    }

    /**
     * 创建一个自定义间隔的sse
     * @param success 成功回调
     * @param error 失败回调
     * @param interval 间隔时间
     * @return sse
     */
    public Flux<String> createSSE(Supplier<String> success, Supplier<String> error, int interval) {
        String account = UserHelper.getNowAccount();
        String uuid = UUID.randomUUID().toString(true);
        return Flux
                .interval(Duration.ofSeconds(interval))
                .doOnSubscribe((a) -> log.info("创建订阅:{} 用户:{} 当前订阅数:{}", uuid, account, current.incrementAndGet()))
                .doOnCancel(() -> log.info("取消订阅:{} 用户:{} 当前订阅数:{}", uuid, account, current.decrementAndGet()))
                .publishOn(Schedulers.fromExecutor(executor))
                // .subscribeOn(Schedulers.fromExecutor(executor))
                // .cancelOn(Schedulers.fromExecutor(executor))
                // .doFinally(signalType -> log.info("订阅结束:{} 类型:{}", uuid, signalType))
                .flatMap(i -> {
                    String current = success.get();
                    // if (i != 0 && StrUtil.isEmpty(current)) {
                    //     return Flux.empty();
                    // }
                    String ret = new String(current.getBytes(StandardCharsets.UTF_8), StandardCharsets.UTF_8);
                    return Flux.just(ret);
                })
                .onErrorResume(Exception.class, e -> {
                    log.error("订阅失败:{}", e.getMessage(), e);
                    return Flux.just(error.get());
                });
    }

}
